/**************************************************************/
/* MTALK.CS */
/* Author: Ohad Rodeh 7/2003 */
/* A simple program implementing a multi-person talk. */
/**************************************************************/


using System;
using System.IO;
using System.Collections;
using System.Net;
using System.Threading;
using Ensemble;
using System.Diagnostics;
using Codebit.ProgramLogger;

/**************************************************************/

public class Mtalk 
{
	static Connection conn = null;
	static Member memb = null;
    static LogFile Log = null;
    static Process currentProcess = System.Diagnostics.Process.GetCurrentProcess();
    static int numOfChkptReadyReplies = 1;
    static int numOfChkptDone = 1;
    static int chkPointNumber = 1;
	
	/**************************************************************/
	
	// A method for the input-thread
	void run () 
	{
        string[] args = Environment.GetCommandLineArgs();
        // Init new TXT-log
        Log = new LogFile("SaS.log", false, LogFile.LogType.TXT, LogFile.LogLevel.All);
        Log.DefaultLogLevel = LogFile.LogLevel.NONE;

        Console.WriteLine("Running and Waiting for Checkpoint...");
        Console.WriteLine("Checkpoint will happen in ~60 seconds...");

		while(true) 
		{
			// Parse an input line and perform the required operation
			//Console.WriteLine("reading from input");
			//string line = Console.ReadLine();
            string line;
            if (args.Length > 1)
                line = ReadLine(Convert.ToInt32(args[1]));
            else
                line = ReadLine(60000);
			lock (conn.send_mutex) 
			{
                if (memb.current_status == Member.Status.Normal)
                {
                    if (line == "chkptReq" && memb.current_view.rank == 0)
                    {
                        Log.Log(System.Environment.NewLine);
                        Log.Log("<COORDINATOR> CHKPOINT NUMBER " + chkPointNumber++);
                        if (memb.current_view.nmembers > 1)
                        {
                            Log.Log("<COORDINATOR> Coordinator is broadcasting chkptReq message to all other processes.");
                            memb.Cast(System.Text.Encoding.ASCII.GetBytes(line));
                        }
                        else
                        {
                            Log.Log("<COORDINATOR> There is only one process (this one) in the view.");
                            Log.Log("<COORDINATOR> Coordinator is taking a local checkpoint...");
                        }


                    }
                    else if(line == "quit")
                    {
                        memb.Leave();
                        //memb.Send1(0, System.Text.Encoding.ASCII.GetBytes(line));
                    }
                }
                else
                    Console.WriteLine("Blocked currently, please try again later");
			}
			Thread.Sleep(100);
		}
	}

    string ReadLine(int timeoutms)
    {
        ReadLineDelegate d = Console.ReadLine;
        IAsyncResult result = d.BeginInvoke(null, null);
        result.AsyncWaitHandle.WaitOne(timeoutms);//timeout e.g. 15000 for 15 secs
        if (result.IsCompleted)
        {
            string resultstr = d.EndInvoke(result);
            //Console.WriteLine("Read: " + resultstr);
            return resultstr;
        }
        else
        {
            //Console.WriteLine("Sending Checkpoint message...");
            return "chkptReq";
        }
    }

    delegate string ReadLineDelegate();

	/**************************************************************/

	static void MainLoop()
	{
		
		// Open a special thread to read from the console
		Mtalk mt = new Mtalk();
		Thread input_thr = new Thread(new ThreadStart(mt.run));
		input_thr.Start();
        string text;
		
		while(true) 
		{
			// Read all waiting messages from Ensmeble
			while (conn.Poll()) 
			{
				Message msg = conn.Recv();
				switch(msg.mtype) 
				{
					case UpType.VIEW:
						//Console.WriteLine("Install:{ " + memb.current_view.name);
						//Console.WriteLine("  nmembers= " + msg.view.nmembers);
						//Console.Write("  view=[");
                        //foreach (string endpt in msg.view.view)
                        //{
                        //    //System.Console.Write(":" + endpt);
                        //}
						//Console.WriteLine("]");
                        //if (writer != null)
                        //{
                        //    writer.Close();
                        //    writer = null;
                        //}
						break;
					case UpType.CAST:
                        if (msg.data == null)
                            text = "";
                        else
                            text = System.Text.Encoding.ASCII.GetString(msg.data);
                        if (text == "chkptReq")
                        {
                            Log.Log("rank<" + memb.current_view.rank + "> Received chkptReq.");
                            Log.Log("rank<" + memb.current_view.rank + "> All the sending messages are received.");
                            Log.Log("rank<" + memb.current_view.rank + "> Sending chkptReady to coordinator.");

                            string replyMsg = "chkptReady";
                            memb.Send1(0, System.Text.Encoding.ASCII.GetBytes(replyMsg));
                        }
                        else if (text == "chkptDo")
                        {
                            Log.Log("rank<" + memb.current_view.rank + "> Received chkptDo.");
                            Log.Log("rank<" + memb.current_view.rank + "> Taking a snapshot....");
                            Log.Log("rank<" + memb.current_view.rank + "> Sending chkptDone to coordinator.");

                            string replyMsg = "chkptDone";
                            memb.Send1(0, System.Text.Encoding.ASCII.GetBytes(replyMsg));
                        }
                        else if (text == "chkptCommit")
                        {
                            Log.Log("rank<" + memb.current_view.rank + "> Received chkptCommit.");
                            Log.Log("rank<" + memb.current_view.rank + "> Resuming the application...");
                        }
                        else
                        {
                            //Console.WriteLine("CAST from (" + msg.origin + ") " + text);
                            //Console.WriteLine("Rank (" + memb.current_view.rank + ")");
                            //Console.WriteLine("Primary (" + memb.current_view.primary + ")");
                        }
						break;
					case UpType.SEND:
                        if (msg.data == null)
                            text = "";
                        else
                            text = System.Text.Encoding.ASCII.GetString(msg.data);

                        if (text == "chkptReady")
                        {
                            //Console.WriteLine("<COORDINATOR> Coordinator received chkptReady from (" + msg.origin + ")");
                            Log.Log("<COORDINATOR> Coordinator received chkptReady from (" + msg.origin + ")");
                            numOfChkptReadyReplies++;
                            if (memb.current_view.nmembers > numOfChkptReadyReplies)
                            {
                                Log.Log("<COORDINATOR> Coordinator is waiting for " + (memb.current_view.nmembers - numOfChkptReadyReplies) + " chkptReady messages.");
                            }
                            else if (memb.current_view.nmembers == numOfChkptReadyReplies)
                            {
                                Log.Log("<COORDINATOR> Coordinator received chkptReady from all processes.");
                                Log.Log("<COORDINATOR> Coordinator is taking a local checkpoint...");
                                Log.Log("<COORDINATOR> Coordinator is broadcasting chkptDo message to all other processes.");
                                memb.Cast(System.Text.Encoding.ASCII.GetBytes("chkptDo"));
                                numOfChkptReadyReplies = 1;
                            }
                            else if (memb.current_view.nmembers < numOfChkptReadyReplies)
                            {
                                //Console.WriteLine("pid<" + currentProcess.Id + "> ASSERT! Two many replies received from (" + msg.origin + ") " + text);
                            }
                        }
                        else if (text == "chkptDone")
                        {
                            Log.Log("<COORDINATOR> Received chkptDone from (" + msg.origin + ")");
                            numOfChkptDone++;
                            if (memb.current_view.nmembers > numOfChkptDone)
                            {
                                Log.Log("<COORDINATOR> Coordinator is waiting for " + (memb.current_view.nmembers - numOfChkptDone) + " chkptDone messages.");
                            }
                            else if (memb.current_view.nmembers == numOfChkptDone)
                            {
                                Log.Log("<COORDINATOR> Received chkptDone from all processes.");
                                Log.Log("<COORDINATOR> Coordinator is broadcasting chkptCommit message to all other processes.");
                                memb.Cast(System.Text.Encoding.ASCII.GetBytes("chkptCommit"));
                                numOfChkptDone = 1;
                            }
                            else if (memb.current_view.nmembers < numOfChkptDone)
                            {
                                //Console.WriteLine("pid<" + currentProcess.Id + "> ASSERT! Two many replies received from (" + msg.origin + ") " + text);
                            }
                        }
						break;
					case UpType.BLOCK:
						memb.BlockOk();
						break;
					case UpType.EXIT:
                        System.Environment.Exit(0);
						break;
				}
				Console.Out.Flush();
			}
			Thread.Sleep(100);
		}
	}

	public static void Main(string[] args)
	{
		conn = new Connection ();
		
		conn.Connect();
		
		JoinOps jops = new JoinOps();
		jops.group_name = "CS_Mtalk" ;

		// Create the endpoint
		memb = new Member(conn);
		
		memb.Join(jops);
		MainLoop();
	}
}

